package com.tqz.rocketmq.transaction;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * <p>
 *
 * </p>
 *
 * @author tian
 * @since 2021/2/13 20:01
 */
public class TransactionListenerLocal implements TransactionListener {

    private Map<String, Boolean> results = new ConcurrentHashMap();

    /**
     * 执行本地事务
     *
     * @param msg
     * @param arg
     * @return
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        System.out.println("开始执行本地事务：" + arg.toString());
        String orderId = arg.toString();
        boolean result = Math.abs(Objects.hash(orderId)) % 2 == 0;
        System.out.println(result);
        results.put(orderId, result);
        return result ? LocalTransactionState.COMMIT_MESSAGE : LocalTransactionState.UNKNOW;
    }

    /**
     * 提供给事务执行状态检查的回调方法，给borker用的(异步回调)
     *
     * @param msg
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        String orderId = msg.getKeys();
        System.out.println("执行事务回调检查：orderId：" + orderId);
        boolean result = results.get(orderId);
        System.out.println("处理结果：" + result);
        return result ? LocalTransactionState.COMMIT_MESSAGE : LocalTransactionState.UNKNOW;
    }
}
